package org.jetlinks.pro.rule.engine.executor;

import com.alibaba.fastjson.JSONObject;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.hswebframework.web.authorization.ReactiveAuthenticationHolder;
import org.jetlinks.core.event.EventBus;
import org.jetlinks.core.event.Subscription;
import org.jetlinks.pro.assets.AssetsHolder;
import org.jetlinks.pro.rule.engine.RuleEngineConstants;
import org.jetlinks.pro.rule.engine.cluster.SchedulerSelectorStrategy;
import org.jetlinks.pro.rule.engine.editor.annotation.EditorResource;
import org.jetlinks.pro.rule.engine.model.nodes.NodeConverter;
import org.jetlinks.pro.tenant.TenantMember;
import org.jetlinks.pro.topic.Topics;
import org.jetlinks.reactor.ql.ReactorQL;
import org.jetlinks.rule.engine.api.RuleConstants;
import org.jetlinks.rule.engine.api.RuleData;
import org.jetlinks.rule.engine.api.RuleDataHelper;
import org.jetlinks.rule.engine.api.model.RuleNodeModel;
import org.jetlinks.rule.engine.api.task.ExecutionContext;
import org.jetlinks.rule.engine.api.task.TaskExecutor;
import org.jetlinks.rule.engine.api.task.TaskExecutorProvider;
import org.jetlinks.rule.engine.defaults.AbstractTaskExecutor;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.*;

@Component
@Slf4j
@EditorResource(
    id = "reactor-ql",
    name = "ReactorQL",
    editor = "rule-engine/editor/function/11-reactor-ql.html",
    helper = "rule-engine/i18n/zh-CN/function/11-reactor-ql.html",
    order = 80
)
@ConfigurationProperties(prefix = "rule.engine.task-executor.reactor-ql")
public class ReactorQLTaskExecutorProvider implements TaskExecutorProvider, NodeConverter {

    private final EventBus eventBus;

    @Getter
    @Setter
    private boolean enableTenant;

    public ReactorQLTaskExecutorProvider(EventBus eventBus) {
        this.eventBus = eventBus;
    }

    @Override
    public String getExecutor() {
        return "reactor-ql";
    }

    @Override
    public Mono<TaskExecutor> createTask(ExecutionContext context) {
        return Mono.just(new ReactorQLTaskExecutor(context));
    }

    @Override
    public String getNodeType() {
        return getExecutor();
    }

    @Override
    public RuleNodeModel convert(JSONObject nodeJson) {
        RuleNodeModel nodeModel = new RuleNodeModel();

        Map<String, Object> config = new HashMap<>();
        config.put("sql", nodeJson.getString("sql"));
        nodeModel.setConfiguration(config);
        return nodeModel;
    }

    class ReactorQLTaskExecutor extends AbstractTaskExecutor {

        private ReactorQL reactorQL;

        public ReactorQLTaskExecutor(ExecutionContext context) {
            super(context);
            init();
        }


        @Override
        public String getName() {
            return "ReactorQL";
        }

        @Override
        protected Disposable doStart() {
            Flux<Object> dataStream;
            //有上游节点
            if (!CollectionUtils.isEmpty(context.getJob().getInputs())) {

                dataStream = context
                    .getInput()
                    .accept()
                    .flatMap(ruleData -> reactorQL
                        .start(Flux.just(RuleDataHelper.toContextMap(ruleData)))
                        .map(ruleData::newData)
                        .onErrorResume(err -> {
                            context.getLogger().error(err.getMessage(), err);
                            return context.onError(err, null).then(Mono.empty());
                        }));
            } else {
                dataStream = reactorQL
                    .start(table -> {
                        if (table == null || table.equalsIgnoreCase("dual")) {
                            return Flux.just(1);
                        }
                        if (table.startsWith("/")) {
                            boolean clusterSubscriber =
                                context.getJob().getSchedulingRule() == null
                                    || SchedulerSelectorStrategy.isAll(context.getJob().getSchedulingRule());
                            return this
                                .refactorTable(table)
                                .flatMapMany(topics -> {
                                    //从事件总线里订阅
                                    return eventBus
                                        .subscribe(Subscription.of(
                                            "rule-engine:"
                                                .concat(context.getInstanceId())
                                                .concat(":")
                                                .concat(context.getJob().getNodeId()),
                                            topics.toArray(new String[0]),
                                            clusterSubscriber
                                                ? new Subscription.Feature[]{Subscription.Feature.local}
                                                : new Subscription.Feature[]{Subscription.Feature.local, Subscription.Feature.broker}
                                        ))
                                        .flatMap(payload -> {
                                            try {
                                                return Mono.just(payload.bodyToJson(true));
                                            } catch (Throwable error) {
                                                return context.onError(error, null);
                                            }
                                        });
                                });
                        }
                        return Flux.just(1);
                    })
                    .cast(Object.class);
            }

            return dataStream
                .flatMap(result -> {
                    RuleData data = context.newRuleData(result);
                    //输出到下一节点
                    return context
                        .getOutput()
                        .write(Mono.just(data))
                        .then(context.fireEvent(RuleConstants.Event.result, data));
                })
                .onErrorResume(err -> context.onError(err, null))
                .subscribe();
        }

        protected Mono<List<String>> refactorTable(String table) {
            String userId = RuleEngineConstants.getCreatorId(context).orElse(null);
            if (isEnableTenant() && StringUtils.hasText(userId)) {
                // TODO: 2021/10/20 处理多个topic可能订阅到同一条消息的问题
                return AssetsHolder
                    .refactorTopic(
                        ReactiveAuthenticationHolder
                            .get(userId)
                            .flatMap(AssetsHolder::fromAuth),
                        table
                    )
                    .collectList()
                    .filter(org.apache.commons.collections4.CollectionUtils::isNotEmpty)
                    .defaultIfEmpty(Collections.singletonList(table))
                    ;
            }
            return Mono.just(Collections.singletonList(table));
        }

        protected ReactorQL createQl() {
            try {

                ReactorQL.Builder builder = Optional
                    .ofNullable(context.getJob().getConfiguration())
                    .map(map -> map.get("sql"))
                    .map(String::valueOf)
                    .map(ReactorQL.builder()::sql)
                    .orElseThrow(() -> new IllegalArgumentException("配置sql错误"));
                return builder.build();
            } catch (Exception e) {
                throw new IllegalArgumentException("SQL格式错误:" + e.getMessage(), e);
            }
        }

        void init() {
            reactorQL = createQl();
        }

        @Override
        public void reload() {
            init();
            if (this.disposable != null) {
                this.disposable.dispose();
            }
            disposable = doStart();
        }

        @Override
        public void validate() {
            createQl();
        }
    }
}
